package com.swak.telnet;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import com.swak.App;
import com.swak.Constants;
import com.swak.OS;
import com.swak.exception.RemotingException;
import com.swak.exception.RpcException;
import com.swak.reactivex.future.Futures;
import com.swak.reactivex.threads.Contexts;
import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.reactivex.transport.ExchangeClient;
import com.swak.reactivex.transport.ExchangeHandler;
import com.swak.reactivex.transport.ExchangeHandlerAdapter;
import com.swak.reactivex.transport.HeaderExchangeClient;
import com.swak.reactivex.transport.HeaderExchangeHandler;
import com.swak.reactivex.transport.HeartbeatHandler;
import com.swak.reactivex.transport.ReferenceCountExchangeClient;
import com.swak.reactivex.transport.TransportMode;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.RegistryService;
import com.swak.registry.URL;
import com.swak.telnet.cmd.CmdInvoker;
import com.swak.telnet.cmd.Command;
import com.swak.telnet.function.CmdFunction;
import com.swak.telnet.function.HandlerFunction;
import com.swak.telnet.function.HintFunction;
import com.swak.telnet.function.RouterFunction;
import com.swak.telnet.invoker.TelnetInvoker;
import com.swak.telnet.transport.TelnetClient;
import com.swak.telnet.transport.TelnetServer;
import com.swak.utils.CollectionUtils;
import com.swak.utils.Ints;
import com.swak.utils.Lists;
import com.swak.utils.Maps;

import lombok.Getter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * 入口: 随机端口号 10000 ~ 20000
 * 
 * @author lifeng
 */
public class Telnet implements DisposableBean, InitializingBean {

	private static Logger logger = LoggerFactory.getLogger(Telnet.class);
	private static Telnet instance;

	/**
	 * 服务器使用一个线程，所有客户端使用一个线程组（无阻塞的调用）
	 */
	private static final LoopResources resources = Contexts.createEventLoopResources(TransportMode.OS, 1, 1,
			"Telnet.EventLoop-", true, 2, TimeUnit.SECONDS);

	public static synchronized Telnet getInstance() {
		if (instance == null) {
			instance = new Telnet();
		}
		return instance;
	}

	/**
	 * 请求处理器
	 */
	private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {

		@Override
		public void connected(Channel channel) throws RemotingException {
			channel.send("Welcome to Telnet! entry help show all command!");
		}

		@Override
		@SuppressWarnings("unchecked")
		public CompletionStage<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
			if (!(message instanceof Command)) {
				throw new RemotingException("Unsupported request: "
						+ (message == null ? null : (message.getClass().getName() + ": " + message))
						+ ", channel: consumer: " + channel.remoteAddress() + " --> provider: "
						+ channel.localAddress());
			}
			Command request = (Command) message;
			HandlerFunction handler = router.route(request);
			if (handler == null) {
				handler = new HintFunction();
			}
			Object result = handler.handle(request);
			if (result instanceof CompletionStage) {
				return (CompletionStage<Object>) result;
			} else if (result instanceof Mono) {
				return ((Mono<Object>) result).toFuture();
			} else if (result instanceof Flux) {
				return (CompletionStage<Object>) Mono.from((Flux<?>) result).toFuture();
			}
			return Futures.future(result);
		}

		@Override
		public void received(Channel channel, Object message) throws RemotingException {
			if (message instanceof Command) {
				this.reply((ExchangeChannel) channel, message);
			} else {
				super.received(channel, message);
			}
		}

		@Override
		public void caught(Channel channel, Throwable t) throws RemotingException {
			logger.error("handle message error：", t);
		}
	};

	private Lock lock = new ReentrantLock();
	private RouterFunction router;
	private TelnetServer server;
	private Map<String, List<ReferenceCountExchangeClient>> clientMaps = Maps.newConcurrentMap();
	private List<TelnetInvoker> invokers = Lists.newArrayList();

	@Getter
	private URL url;
	@Getter
	private List<CmdInvoker> commands;

	@Autowired
	protected App app;

	@Autowired(required = false)
	protected RegistryService registryService;

	private Telnet() {
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		int Telnet_Port = Ints.random(20000, 10000);
		Map<String, String> params = Maps.newHashMap();
		params.put(Constants.Bind_Host, Constants.ANYHOST);
		params.put(Constants.Application_KEY, this.app.getName());
		params.put(Constants.Category_Key, Constants.Registry_Telnet_Server_Key);
		this.url = new URL(Constants.Registry_Telnet_Key, OS.ip(), Telnet_Port, params);
	}

	/**
	 * 返回资源
	 * 
	 * @return
	 */
	public LoopResources resource() {
		return resources;
	}

	/**
	 * 发布服务
	 * 
	 * @param cmds
	 * @throws RemotingException 
	 */
	public void start(Set<Object> cmds) throws RemotingException {
		lock.lock();
		try {

			// 支持的命令
			this.commands = cmds.stream().map(cmd -> CmdInvoker.of(cmd)).collect(Collectors.toList());
			this.router = commands.stream().map(cmd -> CmdFunction.of(cmd)).reduce(RouterFunction::and).orElse(null);
			this.server = new TelnetServer(this.url, resources,
					new HeartbeatHandler(new HeaderExchangeHandler(requestHandler)));

			// 发布服务
			this.export();

			// 显示服务
		} finally {
			lock.unlock();
		}
	}

	@Override
	public void destroy() throws Exception {
		try {
			if (server != null) {
				server.close();
			}
		} catch (Exception e) {
		}
		try {
			resources.dispose();
		} catch (Exception e) {
		}
		try {
			for (TelnetInvoker invoker : this.invokers) {
				invokers.remove(invoker);
				invoker.destroy();
			}
		} catch (Exception e) {
		}
	}

	/**
	 * 异步暴露服务
	 */
	private void export() {
		if (this.registryService != null && this.url != null) {
			this.registryService.register(this.url);
		}
	}

	/**
	 * 创建Telnet 消息发送器
	 * 
	 * @param url
	 * @return
	 */
	public TelnetInvoker newTelnetInvoker(URL url) {
		TelnetInvoker invoker = new TelnetInvoker(url, this.getClients(url));
		this.invokers.add(invoker);
		return invoker;
	}

	private ExchangeClient[] getClients(URL url) {
		List<ReferenceCountExchangeClient> shareClients = getSharedClient(url, 1);
		ExchangeClient[] clients = new ExchangeClient[1];
		for (int i = 0; i < clients.length; i++) {
			clients[i] = shareClients.get(i);
		}
		return clients;
	}

	private List<ReferenceCountExchangeClient> getSharedClient(URL url, int connectNum) {
		String key = url.getAddress();
		List<ReferenceCountExchangeClient> clients = clientMaps.computeIfAbsent(key, (k) -> {
			return Lists.newArrayList();
		});

		// double check
		if (checkClientCanUse(clients)) {
			batchClientRefIncr(clients);
			return clients;
		}

		synchronized (clients) {

			clients = clientMaps.get(key);

			// double check
			if (checkClientCanUse(clients)) {
				batchClientRefIncr(clients);
				return clients;
			}

			// connectNum must be greater than or equal to 1
			connectNum = Math.max(connectNum, 1);

			// If the clients is empty, then the first initialization is
			if (CollectionUtils.isEmpty(clients)) {
				clients = buildReferenceCountExchangeClientList(url, connectNum);
				clientMaps.put(key, clients);

			} else {
				for (int i = 0; i < clients.size(); i++) {
					ReferenceCountExchangeClient referenceCountExchangeClient = clients.get(i);
					// If there is a client in the list that is no longer available, create a new
					// one to replace him.
					if (referenceCountExchangeClient == null || referenceCountExchangeClient.isClosed()) {
						clients.set(i, buildReferenceCountExchangeClient(url));
						continue;
					}

					referenceCountExchangeClient.incrementAndGetCount();
				}
			}

			return clients;
		}
	}

	private boolean checkClientCanUse(List<ReferenceCountExchangeClient> referenceCountExchangeClients) {
		if (CollectionUtils.isEmpty(referenceCountExchangeClients)) {
			return false;
		}

		for (ReferenceCountExchangeClient referenceCountExchangeClient : referenceCountExchangeClients) {
			// As long as one client is not available, you need to replace the unavailable
			// client with the available one.
			if (referenceCountExchangeClient == null || referenceCountExchangeClient.isClosed()) {
				return false;
			}
		}

		return true;
	}

	/**
	 * Bulk build client
	 *
	 * @param url
	 * @param connectNum
	 * @return
	 */
	private List<ReferenceCountExchangeClient> buildReferenceCountExchangeClientList(URL url, int connectNum) {
		List<ReferenceCountExchangeClient> clients = new ArrayList<>();

		for (int i = 0; i < connectNum; i++) {
			clients.add(buildReferenceCountExchangeClient(url));
		}

		return clients;
	}

	/**
	 * Increase the reference Count if we create new invoker shares same connection,
	 * the connection will be closed without any reference.
	 *
	 * @param referenceCountExchangeClients
	 */
	private void batchClientRefIncr(List<ReferenceCountExchangeClient> referenceCountExchangeClients) {
		if (CollectionUtils.isEmpty(referenceCountExchangeClients)) {
			return;
		}

		for (ReferenceCountExchangeClient referenceCountExchangeClient : referenceCountExchangeClients) {
			if (referenceCountExchangeClient != null) {
				referenceCountExchangeClient.incrementAndGetCount();
			}
		}
	}

	/**
	 * Build a single client
	 *
	 * @param url
	 * @return
	 */
	private ReferenceCountExchangeClient buildReferenceCountExchangeClient(URL url) {
		ExchangeClient exchangeClient = initClient(url);
		return new ReferenceCountExchangeClient(exchangeClient);
	}

	private ExchangeClient initClient(URL url) {
		url = url.addParameter(Constants.Heartbeat, String.valueOf(Constants.Default_Heartbeat));
		try {
			return new HeaderExchangeClient(
					new TelnetClient(url, resources, new HeartbeatHandler(new HeaderExchangeHandler(requestHandler))));
		} catch (Exception e) {
			throw new RpcException(RpcException.UNKNOWN_EXCEPTION,
					"Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
		}
	}
}
